{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "7da3a059",
   "metadata": {},
   "source": [
    "## 串行、并行与并发"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "41bc7e6d",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/1.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2e11b48b",
   "metadata": {},
   "source": [
    "1. 串行(serial)：一个CPU上，按顺序完成多个任务\n",
    "\n",
    "2. 并行(parallelism)：指的是任务数小于等于cpu核数，多个任务在同一时刻 同时运行。\n",
    "3. 并发(concurrency)：一个CPU采用时间片管理方式，交替的处理多个任务。一般是是任务数大于cpu核数，通过操作系统的各种任务调度算法，实现用多个任务“一起”执行（因为切换任务的速度相当快，看上去一起执行而已）"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e7801df1",
   "metadata": {},
   "source": [
    "## 进程、线程、协程"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6baaf180",
   "metadata": {},
   "source": [
    "### 故事引入"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1f3b292b",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/2.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c45b74f0",
   "metadata": {},
   "source": [
    "许家印 想开工厂生产手机，他引进了一条生产线，这个生产线上有很多的器件以及材料，**一条生产线就是一个进程**。\n",
    "\n",
    "只有生产线是不够的，所以找五个工人来进行生产，这个工人能够利用这些材料最终一步步的将手机做出来，**这五个工人就是五个线程**。\n",
    "\n",
    "\n",
    "- 为了提高生产率，许家印 想到3种办法：\n",
    "    1. 一条生产线上多招些工人，一起来做手机，这样效率是成倍増长，即 单进程多线程方式\n",
    "    2. 多条生产线，每个生产线上多个工人，即 多进程多线程\n",
    "\n",
    "\n",
    "- 许家印 深入一线发现工人并不是那么忙，有很多等待时间。于是规定：\n",
    "    - 员工在等待生产线上某个零件时，不要闲着，干点其他工作。\n",
    "    - 也就是说：如果一个线程等待某些事件时，可以充分利用这个时间去做其它事情，这就是：**协程方式**。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8ebf1b50",
   "metadata": {},
   "source": [
    "### 三者关系"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6b13813f",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/3.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ff647920",
   "metadata": {},
   "source": [
    "1. 线程是程序执行的最小单位，而进程是操作系统分配资源的最小单位；\n",
    "\n",
    "2. 一个进程由一个或多个线程组成，线程是一个进程中代码的不同执行路线；\n",
    "\n",
    "3. 进程之间相互独立，但同一进程下的各个线程之间共享程序的内存空间(包括代码段、数据集、堆等)及一些进程级的资源(如打开文件和信号)，某进程内的线程在其它进程不可见；\n",
    "\n",
    "4. 调度和切换：线程上下文切换比进程上下文切换要快得多。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5d468316",
   "metadata": {},
   "source": [
    "- 进程(Process)：拥有自己独立的堆和栈，既不共享堆，也不共享栈，进程由操作系统调度；进程切换需要的资源很最大，效率低\n",
    "\n",
    "- 线程(Thread)：拥有自己独立的栈和共享的堆，共享堆，不共享栈，标准线程由操作系统调度；线程切换需要的资源一般，效率一般\n",
    "\n",
    "- 协程(coroutine)：拥有自己独立的栈和共享的堆，共享堆，不共享栈，协程由程序员在协程的代码里显示调度；协程切换任务资源很小，效率高"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "79a4f9c6",
   "metadata": {},
   "source": [
    "### 进程"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b7ed1d26",
   "metadata": {},
   "source": [
    "进程（Process）是一个 具有一定独立功能的程序 关于某个数据集合的 一次运行活动。\n",
    "\n",
    "现代操作系统比如 Mac OS X，Linux，Windows等，都是支持“多任务”的操作系统，什么叫“多任务”呢？\n",
    "- 简单地说，就是操作系统可以同时运行多个任务。\n",
    "- 打个比方，你一边在用逛淘宝，一边在听音乐，一边在用微信聊天，这就是多任务，至少同时有3个任务正在运行。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "18a9c1e3",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/4.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "30be4336",
   "metadata": {},
   "source": [
    "### 线程\n",
    "线程（Thread）是操作系统能够进行运算调度的最小单位。它被包含在进程之中，是进程中的实际运作单位。\n",
    "\n",
    "有些进程还不止同时干一件事，比如微信，它可以同时进行打字聊天，视频聊天，朋友圈等事情。\n",
    "\n",
    "在一个进程内部，要同时干多件事，就需要同时运行多个“子任务”，我们把进程内的这些“子任务”称为线程（Thread）。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "afd44f2c",
   "metadata": {},
   "source": [
    "**并发编程解决方案：**\n",
    "1. 多进程模式：启动多个进程，每个进程虽然只有一个线程，但多个进程可以一块执行多个任务\n",
    "1. 多线程模式：启动一个进程，在一个进程内启动多个线程，这样，多个线程也可以一块执行多个任务\n",
    "1. 多进程+多线程模式：启动多个进程，每个进程再启动多个线程，这样同时执行的任务就更多了"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "98cf8d28",
   "metadata": {},
   "source": [
    "### 协程\n",
    "协程（Coroutines），也叫作纤程(Fiber)，是一种比线程更加轻量级的存在，由程序员自己写程序来管理。\n",
    "\n",
    "当出现IO阻塞时，CPU一直等待IO返回，处于空转状态。这时候用协程，可以执行其他任务。\n",
    "\n",
    "当IO返回结果后，再回来处理数据。充分利用了IO等待的时间，提高了效率。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2f4dbadd",
   "metadata": {},
   "source": [
    "## 同步和异步"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3db64faf",
   "metadata": {},
   "source": [
    "同步和异步强调的是消息通信机制 (synchronous communication / asynchronous communication)\n",
    "\n",
    "\n",
    "**同步( synchronous )**：A调用B，等待B返回结果后，A继续执行。\n",
    "\n",
    "\n",
    "**异步( asynchronous )**：A调用B，A继续执行，不等待B返回结果；B有结果了，通知A，A再做处理。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6c87da44",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/5.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dee57861",
   "metadata": {},
   "source": [
    "## Python与线程Threading"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4185dcf7",
   "metadata": {},
   "source": [
    "### 方法模式 创建线程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "72ae8f5f",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-06T07:06:43.814527Z",
     "start_time": "2023-10-06T07:06:43.809877Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "主线程，start\n",
      "线程t1, start线程t2, start\n",
      "线程t1, 0\n",
      "主线程,end\n",
      "\n",
      "线程t2, 0\n"
     ]
    }
   ],
   "source": [
    "# 导包\n",
    "from threading import Thread\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "def func1(name):\n",
    "    print(f\"线程{name}, start\")\n",
    "    \n",
    "    for i in range(3):\n",
    "        \n",
    "        # A线程在运行的时候，cpu资源可能会被其他线程抢占，使得 换行符 没来得及打印\n",
    "        print(f\"线程{name}, {i}\")\n",
    "        \n",
    "        # 暂停1秒\n",
    "        sleep(1)\n",
    "        \n",
    "    print(f\"线程{name}, end\")\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    print(\"主线程，start\")\n",
    "    \n",
    "    # 创建线程1\n",
    "    t1 = Thread(target=func1, # 线程执行的函数\n",
    "                args=(\"t1\",)) # 给执行函数传递 参数 -- 必须是元组类型\n",
    "    \n",
    "    # 创建线程2\n",
    "    t2 = Thread(target=func1, args=(\"t2\",))\n",
    "    \n",
    "    # 启动线程\n",
    "    t1.start()\n",
    "    t2.start()\n",
    "    \n",
    "    print(\"主线程,end\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "22ac42e2",
   "metadata": {},
   "source": [
    "### 类包装 创建线程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "08da5d35",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-06T07:06:43.820968Z",
     "start_time": "2023-10-06T07:06:43.815730Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "主线程，start\n",
      "线程t1, start线程t2, start\n",
      "线程t1, 0\n",
      "\n",
      "线程t2, 0\n",
      "主线程, end\n"
     ]
    }
   ],
   "source": [
    "from threading import Thread\n",
    "from time import sleep\n",
    "\n",
    "class MyThread(Thread):\n",
    "    \"\"\"继承 线程Thread类\"\"\"\n",
    "    \n",
    "    def __init__(self,name):\n",
    "        # 调用父类初始化方法\n",
    "        Thread.__init__(self)\n",
    "        self.name = name\n",
    "\n",
    "    def run(self):\n",
    "        \"\"\"重写 run方法：启动线程时，自动执行run方法\"\"\"\n",
    "        \n",
    "        print(f\"线程{self.name}, start\")\n",
    "        \n",
    "        for i in range(3):\n",
    "            \n",
    "             # A线程在运行的时候，cpu资源可能会被其他线程抢占，使得 换行符 没来得及打印\n",
    "            print(f\"线程{self.name}, {i}\")\n",
    "            \n",
    "            sleep(1)\n",
    "            \n",
    "        print(f\"线程{self.name}, end\")\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    print(\"主线程，start\")\n",
    "    \n",
    "    #创建线程\n",
    "    t1 = MyThread(\"t1\")\n",
    "    t2 = MyThread(\"t2\")\n",
    "    \n",
    "    #启动线程\n",
    "    t1.start()\n",
    "    t2.start()\n",
    "    \n",
    "    print(\"主线程, end\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9d9f6988",
   "metadata": {},
   "source": [
    "### join方法\n",
    "使用 join() 方法\n",
    "\n",
    "之前的代码，主线程不会等待子线程结束，主线程会自顾自地提前运行结束 -- 异步\n",
    "\n",
    "如果需要等待子线程结束后，主线程再往下执行，可使用join()方法 -- 同步"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "e6f637b6",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-06T07:06:46.845056Z",
     "start_time": "2023-10-06T07:06:43.824248Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "主线程，start\n",
      "thread:t1, 0\n",
      "thread:t2, 0\n",
      "线程t1, 1线程t2, 1\n",
      "\n",
      "线程t1, 1\n",
      "线程t2, 1\n",
      "thread:t1, 1thread:t2, 1\n",
      "\n",
      "线程t1, 2\n",
      "线程t2, 2线程t1, 2\n",
      "\n",
      "线程t2, 2\n",
      "thread:t1, 2thread:t2, 2\n",
      "\n",
      "线程t1, end线程t2, end线程t1, end线程t2, end\n",
      "\n",
      "\n",
      "\n",
      "主线程，end\n"
     ]
    }
   ],
   "source": [
    "# 导包\n",
    "from threading import Thread\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "def func1(name):\n",
    "    for i in range(3):\n",
    "        print(f\"thread:{name}, {i}\")\n",
    "        sleep(1)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    print(\"主线程，start\")\n",
    "    \n",
    "    # 创建线程\n",
    "    t1 = Thread(target=func1, args=(\"t1\",))\n",
    "    t2 = Thread(target=func1, args=(\"t2\",))\n",
    "    \n",
    "    # 启动线程\n",
    "    t1.start()\n",
    "    t2.start()\n",
    "    \n",
    "    # 主线程会等待t1, t2结束后，再往下执行\n",
    "    t1.join()\n",
    "    t2.join()\n",
    "    \n",
    "    print(\"主线程，end\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "77cf0cf3",
   "metadata": {},
   "source": [
    "### 守护线程\n",
    "守护线程 主要的特征是它的生命周期。主线程死亡，它也就随之死亡。\n",
    "\n",
    "在python中，线程通过修改 daemon 属性 来设置是否为守护线程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "2be021bd",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-06T07:06:46.857954Z",
     "start_time": "2023-10-06T07:06:46.848733Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "主线程，start\n",
      "thread:t1, 0\n",
      "thread:t2, 0主线程，end\n",
      "\n"
     ]
    }
   ],
   "source": [
    "\"\"\"注：不要守护线程和非守护线程同时存在，不然容易出现问题\"\"\"\n",
    "\n",
    "from threading import Thread\n",
    "from time import sleep\n",
    "\n",
    "class MyThread(Thread):\n",
    "    \n",
    "    def __init__(self, name, daemon=False):\n",
    "        Thread.__init__(self)\n",
    "        Thread.daemon = daemon\n",
    "        self.name =name\n",
    "        \n",
    "    def run(self):\n",
    "        for i in range(3):\n",
    "            print(f\"thread:{self.name}, {i}\")\n",
    "            sleep(1)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    print(\"主线程，start\")\n",
    "    \n",
    "    # 创建线程(类的方式)\n",
    "    t1 = MyThread('t1')\n",
    "    t2 = MyThread('t2', daemon=True) # 直接定义 t2为守护线程\n",
    "    \n",
    "    # t1设置为守护线程\n",
    "    t1.daemon = True\n",
    "    # 另一种方式：t1.setDaemon(True)\n",
    "    \n",
    "    # 启动线程\n",
    "    t1.start()\n",
    "    t2.start()\n",
    "    \n",
    "    print(\"主线程，end\")\n",
    "    \n",
    "    # 在jupyter notebook上，要想得到预期效果，加上exit()\n",
    "    exit()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "325c7cda",
   "metadata": {},
   "source": [
    "### 全局锁GIL问题"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ab937bf0",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/6.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "30fe6389",
   "metadata": {},
   "source": [
    "在python中，无论你电脑有多少核，同一时间执行的线程只有一个线程。这个是python的一个开发时候，设计的一个缺陷，所以说python中的线程是“含有水分的线程”。\n",
    "\n",
    "**Python GIL(Global Interpreter Lock)：**\n",
    "- Python代码的执行由Python 虚拟机(也叫解释器主循环，CPython版本)来控制\n",
    "- Python 在设计之初就考虑到要在解释器的主循环中，同时只有一个线程在执行，即在任意时刻，只有一个线程在解释器中运行。\n",
    "- 对Python 虚拟机的访问由全局解释器锁（GIL）来控制，正是这个锁能保证同一时刻只有一个线程在运行。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "bf87b8eb",
   "metadata": {},
   "source": [
    "### 线程同步\n",
    "同一个资源，多人想用？排队……\n",
    "\n",
    "![](01_并发编程_images/7.png)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "58c07506",
   "metadata": {},
   "source": [
    "**线程同步的概念：**\n",
    "- 处理多线程问题时，多个线程访问同一个对象，并且某些线程还想修改这个对象。 这时候，我们就需要用到“线程同步”。 \n",
    "- 线程同步其实就是一种等待机制，多个需要同时访问此对象的线程进入这个对象的等待池形成队列，等待前面的线程使用完毕后，下一个线程再使用。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d1744918",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/8.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "50b0dfec",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-06T08:25:58.440514Z",
     "start_time": "2023-10-06T08:25:58.431769Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "账户：zxy,余额是：20\n",
      "账户：zxy,总共取了：80\n",
      "账户：zxy,余额是：-60\n",
      "账户：zxy,总共取了：80\n"
     ]
    }
   ],
   "source": [
    "\"\"\"未使用线程同步和互斥锁的情况\"\"\"\n",
    "from threading import Thread\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "class Account:\n",
    "    \"\"\"账户\"\"\"\n",
    "    def __init__(self, money, name):\n",
    "        self.money = money\n",
    "        self.name = name\n",
    "\n",
    "\n",
    "class Drawing(Thread):\n",
    "    \"\"\"模拟提款的操作\"\"\"\n",
    "    def __init__(self, drawingNum, account):\n",
    "        Thread.__init__(self)\n",
    "        \n",
    "        # 提款金额\n",
    "        self.drawingNum = drawingNum\n",
    "        \n",
    "        self.account = account\n",
    "        \n",
    "        # 总提款数\n",
    "        self.expenseTotal = 0\n",
    "\n",
    "    def run(self):\n",
    "        \n",
    "        # 账户余额不足，则退出\n",
    "        if self.account.money < self.drawingNum:\n",
    "            return\n",
    "        \n",
    "        # 如果可以取钱，则阻塞\n",
    "        sleep(1)  \n",
    "        \n",
    "        self.account.money -= self.drawingNum\n",
    "        self.expenseTotal += self.drawingNum\n",
    "        \n",
    "        print(f\"账户：{self.account.name},余额是：{self.account.money}\")\n",
    "        \n",
    "        print(f\"账户：{self.account.name},总共取了：{self.expenseTotal}\")\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    \n",
    "    # 创建100元的账户\n",
    "    a1 = Account(100, \"zxy\")\n",
    "    \n",
    "    # 定义两个取钱线程\n",
    "    draw1 = Drawing(80, a1)  \n",
    "    draw2 = Drawing(80, a1)  \n",
    "    \n",
    "    # 运行线程\n",
    "    draw1.start()\n",
    "    draw2.start()\n",
    "    \n",
    "    \"\"\"我们多取了钱，银行肯定不答应\"\"\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "434639d4",
   "metadata": {},
   "source": [
    "### 互斥锁"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8e27dad0",
   "metadata": {},
   "source": [
    "**我们可以通过 “锁机制” 来实现线程同步问题，锁机制有如下几个要点：**\n",
    "1. 必须使用同一个锁对象\n",
    "\n",
    "2. 互斥锁的作用就是保证同一时刻只能有一个线程去操作共享数据，保证共享数据不会出现错误问题\n",
    "\n",
    "3. 使用互斥锁的好处确保某段关键代码只能由一个线程从头到尾完整地去执行\n",
    "\n",
    "4. 使用互斥锁会影响代码的执行效率\n",
    "\n",
    "5. 同时持有多把锁，容易出现死锁的情况\n",
    "\n",
    "\n",
    "**互斥锁是什么？**\n",
    "- 互斥锁: 对共享数据进行锁定，保证同一时刻只能有一个线程去操作。\n",
    "\n",
    "- 注意: 互斥锁是多个线程一起去抢，抢到锁的线程先执行，没有抢到锁的线程需要等待，等互斥锁使用完释放后，其它等待的线程再去抢这个锁。\n",
    "\n",
    "- threading 模块中定义了 Lock 变量，这个变量本质上是一个函数，通过调用这个函数可以获取一把互斥锁。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "ee6db448",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-06T08:33:07.884075Z",
     "start_time": "2023-10-06T08:33:07.875148Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "账户：zxy, 余额是：20账户余额不足！\n",
      "\n",
      "账户：zxy, 总共取了：80\n"
     ]
    }
   ],
   "source": [
    "\"\"\"使用互斥锁的案例\"\"\"\n",
    "from threading import Thread, Lock\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "class Account:\n",
    "    \"\"\"账户\"\"\"\n",
    "    def __init__(self, money, name):\n",
    "        self.money = money\n",
    "        self.name = name\n",
    "\n",
    "\n",
    "class Drawing(Thread):\n",
    "    \"\"\"模拟提款的操作\"\"\"\n",
    "    def __init__(self, drawingNum, account):\n",
    "        Thread.__init__(self)\n",
    "        self.drawingNum = drawingNum\n",
    "        self.account = account\n",
    "        self.expenseTotal = 0\n",
    "\n",
    "    def run(self):\n",
    "        # 获得互斥锁\n",
    "        lock1.acquire()\n",
    "        \n",
    "        if self.account.money < self.drawingNum:\n",
    "            print(\"账户余额不足！\")\n",
    "            return\n",
    "        \n",
    "        # 如果可以取钱，则阻塞。\n",
    "        sleep(1)  \n",
    "        \n",
    "        # 修改相应信息\n",
    "        self.account.money -= self.drawingNum\n",
    "        self.expenseTotal += self.drawingNum\n",
    "        \n",
    "        # 取完钱后，释放互斥锁\n",
    "        lock1.release()\n",
    "        \n",
    "        # 打印信息\n",
    "        print(f\"账户：{self.account.name}, 余额是：{self.account.money}\")\n",
    "        print(f\"账户：{self.account.name}, 总共取了：{self.expenseTotal}\")\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # 创建银行账户\n",
    "    a1 = Account(100, \"zxy\")\n",
    "    \n",
    "    # 创建互斥锁\n",
    "    lock1 = Lock()\n",
    "    \n",
    "    # 创建两个取款 线程\n",
    "    draw1 = Drawing(80, a1)  \n",
    "    draw2 = Drawing(80, a1)  \n",
    "    \n",
    "    # 运行线程\n",
    "    draw1.start()\n",
    "    draw2.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "7349ce1b",
   "metadata": {},
   "source": [
    "### 死锁"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c87f071f",
   "metadata": {},
   "source": [
    "在多线程程序中，死锁问题很大一部分是由于一个线程同时获取多个锁造成的。\n",
    "\n",
    "举例：有两个人都要做饭，都需要“锅”和“菜刀”才能炒菜。\n",
    "\n",
    "![](01_并发编程_images/9.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "7d98046d",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-06T08:37:47.887897Z",
     "start_time": "2023-10-06T08:37:47.879576Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "fun1拿到菜刀\n",
      "fun2拿到锅\n"
     ]
    }
   ],
   "source": [
    "\"\"\"死锁案例\"\"\"\n",
    "from threading import Thread, Lock\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "def fun1():\n",
    "    lock1.acquire()\n",
    "    print('fun1拿到菜刀')\n",
    "    sleep(2)\n",
    "    \n",
    "    lock2.acquire()\n",
    "    print('fun1拿到锅')\n",
    "\n",
    "    print('fun1开始炒菜...')\n",
    "    \n",
    "    lock2.release()\n",
    "    print('fun1释放锅')\n",
    "    \n",
    "    lock1.release()\n",
    "    print('fun1释放菜刀')\n",
    "\n",
    "\n",
    "def fun2():\n",
    "    lock2.acquire()\n",
    "    print('fun2拿到锅')\n",
    "    \n",
    "    lock1.acquire()\n",
    "    print('fun2拿到菜刀')\n",
    "    \n",
    "    print('fun2开始炒菜...')\n",
    "    \n",
    "    lock1.release()\n",
    "    print('fun2释放菜刀')\n",
    "    \n",
    "    lock2.release()\n",
    "    print('fun2释放锅')\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # 创建两个锁\n",
    "    lock1 = Lock()\n",
    "    lock2 = Lock()\n",
    "\n",
    "    # 创建两个线程\n",
    "    t1 = Thread(target=fun1)\n",
    "    t2 = Thread(target=fun2)\n",
    "    \n",
    "    # 运行线程 -- 发生死锁，两个线程都运行不下去，互相等待对方释放资源\n",
    "    t1.start()\n",
    "    t2.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "fc7a9d8a",
   "metadata": {},
   "source": [
    "**死锁的解决方法：**\n",
    "\n",
    "死锁是由于“同步块需要同时持有多个锁造成”的，要解决这个问题，可以规定：同一个代码块，不要同时持有两个对象锁"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a7a637b6",
   "metadata": {},
   "source": [
    "### 信号量 (Semaphore)\n",
    "互斥锁使用后，一个资源同时只有一个线程访问。\n",
    "\n",
    "如果某个资源，我们同时想让N个(指定数值)线程访问，就可以使用信号量。\n",
    "\n",
    "信号量控制同时访问资源的数量。信号量和锁相似，锁同一时间只允许一个对象(进程)通过，信号量同一时间允许多个对象(进程)通过\n",
    "\n",
    "信号量底层就是一个内置的计数器。每当资源获取时(调用acquire)，计数器-1，资源释放时(调用release)计数器+1。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d40541eb",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/10.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "4d385596",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-06T08:46:09.125197Z",
     "start_time": "2023-10-06T08:46:09.118681Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "tom0进入房间\n",
      "tom1进入房间\n",
      "*tom1走出房间\n",
      "tom2进入房间*tom0走出房间\n",
      "\n",
      "tom3进入房间\n",
      "*tom3走出房间\n",
      "*tom2走出房间\n",
      "tom5进入房间tom4进入房间\n",
      "\n",
      "*tom4走出房间\n",
      "tom6进入房间*tom5走出房间\n",
      "\n",
      "*tom6走出房间\n"
     ]
    }
   ],
   "source": [
    "\"\"\"信号量案例：一个房间，只允许两个人进来\"\"\"\n",
    "from threading import Semaphore, Thread\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "def home(name, se):\n",
    "    \"\"\"房间：传入信号量\"\"\"\n",
    "    \n",
    "    # 获取资源\n",
    "    se.acquire()\n",
    "    \n",
    "    print(f\"{name}进入房间\")\n",
    "    sleep(3)\n",
    "    print(f\"*{name}走出房间\")\n",
    "    \n",
    "    # 释放资源\n",
    "    se.release()\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # 信号量对象\n",
    "    se = Semaphore(2)  \n",
    "    \n",
    "    # 依次创建7个线程\n",
    "    for i in range(7):\n",
    "        t = Thread(target=home, args=(f\"tom{i}\", se))\n",
    "        t.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5fc77d39",
   "metadata": {},
   "source": [
    "### 事件(Event)\n",
    "事件Event主要用于唤醒正在阻塞等待状态的线程;\n",
    "\n",
    "![](01_并发编程_images/11.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8a33a7d4",
   "metadata": {},
   "source": [
    "Event 对象包含一个可由线程设置的信号标志，它允许线程等待某些事件的发生。\n",
    "\n",
    "1. 在初始情况下，event 对象中的信号标志被设置假。\n",
    "\n",
    "2. 如果有线程等待一个 event 对象，而这个 event 对象的标志为假，那么这个线程将会被一直阻塞直至该标志为真。\n",
    "\n",
    "3. 一个线程如果将一个 event 对象的信号标志设置为真，它将唤醒所有等待 event 对象的线程。\n",
    "\n",
    "4. 如果一个线程等待一个已经被设置为真的 event 对象，那么它将忽略这个事件，继续执行\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f6dbde99",
   "metadata": {},
   "source": [
    "Event() 可以创建一个事件管理标志，该标志（event）默认为 False，event对象主要有四种方法可以调用：\n",
    "\n",
    "![](01_并发编程_images/12.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "e0fc8a16",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-06T08:57:24.713066Z",
     "start_time": "2023-10-06T08:57:19.694335Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "小伙伴tom已经进入就餐状态！\n",
      "小伙伴cherry已经进入就餐状态！\n",
      "->主线程通知小伙伴开吃！\n",
      "小伙伴cherry收到通知，开吃！\n",
      "小伙伴tom收到通知，开吃！\n"
     ]
    }
   ],
   "source": [
    "\"\"\"\n",
    "小伙伴a,b,c围着吃火锅，当菜上齐了，请客的主人说：开吃！\n",
    "于是小伙伴一起动筷子，这种场景如何实现?\n",
    "\"\"\"\n",
    "import threading\n",
    "import time\n",
    "\n",
    "\n",
    "def hot_pot(name):\n",
    "    \"\"\"吃火锅\"\"\"\n",
    "    \n",
    "    print(f'小伙伴{name}已经进入就餐状态！')\n",
    "    time.sleep(1)\n",
    "    \n",
    "    # 等待事件\n",
    "    event.wait()\n",
    "    \n",
    "    # 收到事件后，进入运行状态\n",
    "    print(f'小伙伴{name}收到通知，开吃！')\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    \n",
    "    # 创建事件对象\n",
    "    event = threading.Event()\n",
    "    \n",
    "    # 创建新线程\n",
    "    thread1 = threading.Thread(target=hot_pot, args=(\"tom\", ))\n",
    "    thread2 = threading.Thread(target=hot_pot, args=(\"cherry\", ))\n",
    "    \n",
    "    # 开启线程\n",
    "    thread1.start()\n",
    "    thread2.start()\n",
    "\n",
    "    time.sleep(5)\n",
    "    \n",
    "    # 发送事件通知\n",
    "    print('->主线程通知小伙伴开吃！')\n",
    "    event.set()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b96a8be6",
   "metadata": {},
   "source": [
    "### 生产者和消费者模式"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ffdc4de5",
   "metadata": {},
   "source": [
    "多线程环境下，我们经常需要多个线程的并发和协作。这个时候，就需要了解一个重要的多线程并发协作模型“生产者/消费者模式”。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "34fd9fcc",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/13.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9bda4ac7",
   "metadata": {},
   "source": [
    "- 生产者指的是负责生产数据的模块（这里模块可能是：方法、对象、线程、进程）\n",
    "\n",
    "- 消费者指的是负责处理数据的模块（这里模块可能是：方法、对象、线程、进程）\n",
    "\n",
    "- 消费者不能直接使用生产者的数据，它们之间有个“缓冲区”。生产者将生产好的数据放入“缓冲区”，消费者从“缓冲区”拿要处理的数据"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "94613c53",
   "metadata": {},
   "source": [
    "**缓冲区和queue对象:**\n",
    "\n",
    "- 从一个线程向另一个线程发送数据最安全的方式可能就是使用queue 库中的队列了。\n",
    "- 创建一个被多个线程共享的 Queue 对象，这些线程通过使用 put() 和 get() 操作来向队列中添加或者删除元素。\n",
    "- Queue 对象已经包含了必要的锁，所以你可以通过它在多个线程间多安全地共享数据。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "b1dc3411",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-08T01:56:36.791264Z",
     "start_time": "2023-10-08T01:56:36.778567Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "生产1号，大馒头\n",
      "获取馒头：大馒头：1号\n",
      "生产2号，大馒头\n",
      "获取馒头：大馒头：2号\n",
      "生产3号，大馒头\n",
      "获取馒头：大馒头：3号\n",
      "生产4号，大馒头\n",
      "获取馒头：大馒头：4号\n",
      "生产5号，大馒头\n",
      "获取馒头：大馒头：5号\n",
      "生产6号，大馒头\n",
      "获取馒头：大馒头：6号\n",
      "生产7号，大馒头\n",
      "获取馒头：大馒头：7号\n",
      "生产8号，大馒头\n",
      "获取馒头：大馒头：8号\n",
      "生产9号，大馒头\n",
      "获取馒头：大馒头：9号\n",
      "生产10号，大馒头\n",
      "获取馒头：大馒头：10号\n",
      "生产11号，大馒头\n",
      "获取馒头：大馒头：11号\n",
      "生产12号，大馒头\n",
      "获取馒头：大馒头：12号\n",
      "生产13号，大馒头\n",
      "获取馒头：大馒头：13号\n",
      "生产14号，大馒头\n",
      "获取馒头：大馒头：14号\n",
      "生产15号，大馒头\n",
      "获取馒头：大馒头：15号\n"
     ]
    }
   ],
   "source": [
    "\"\"\"生产者和消费者模式\"\"\"\n",
    "from queue import Queue\n",
    "from threading import Thread\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "def producer():\n",
    "    \"\"\"生产者\"\"\"\n",
    "    \n",
    "    # 记录馒头数量\n",
    "    num = 1\n",
    "    while True:\n",
    "        if queue.qsize() < 5:\n",
    "            \"\"\"队列最多容纳5个馒头\"\"\"\n",
    "            print(f\"生产{num}号，大馒头\")\n",
    "            queue.put(f\"大馒头：{num}号\")\n",
    "            num += 1\n",
    "        else:\n",
    "            print(\"馒头框满了，等待来人消费啊！\")\n",
    "        sleep(1)\n",
    "\n",
    "\n",
    "def consumer():\n",
    "    \"\"\"消费者\"\"\"\n",
    "    while True:\n",
    "        print(f\"获取馒头：{queue.get()}\")\n",
    "        sleep(1)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    \n",
    "    # 创建 队列对象\n",
    "    queue = Queue()\n",
    "    \n",
    "    # 创造 生产者线程 和 消费者线程\n",
    "    t1 = Thread(target=producer)\n",
    "    t2 = Thread(target=consumer)\n",
    "    \n",
    "    # 启动线程\n",
    "    t1.start()\n",
    "    t2.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9610044c",
   "metadata": {},
   "source": [
    "## Python与进程Processing"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d8946cd0",
   "metadata": {},
   "source": [
    "**进程的优点：**\n",
    "1. 可以使用计算机多核，进行任务的并行执行，提高执行效率\n",
    "2. 运行不受其他进程影响，创建方便\n",
    "3. 空间独立，数据安全\n",
    "\n",
    "**进程的缺点：**\n",
    "1. 进程的创建和删除消耗的系统资源较多"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b5e2050d",
   "metadata": {},
   "source": [
    "### 方法模式 创建进程"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c44209d2",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "下述代码在 jupyter notebook上 无法正常运行，\n",
    "请使用 Pycharm运行，\n",
    "或者将代码保存为.py文件，再导入jupyter运行\n",
    "\"\"\"\n",
    "import os\n",
    "from multiprocessing import Process\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "def fun1(name):\n",
    "    print(f\"当前进程ID:{os.getpid()}\")\n",
    "\n",
    "    # 创建该进程的 进程，这里为主进程\n",
    "    print(f\"父进程ID：{os.getppid()}\")\n",
    "\n",
    "    print(f\"Process:{name}, start\")\n",
    "\n",
    "    sleep(1)\n",
    "\n",
    "    print(f\"Process:{name}, end\")\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "\n",
    "    print(\"当前进程ID:\", os.getpid())\n",
    "\n",
    "    # 创建进程\n",
    "    p1 = Process(target=fun1,  # 进程负责执行的函数\n",
    "                 args=(\"p1\",))  # 函数内传入的参数 -- 必须是元组类型\n",
    "\n",
    "    p2 = Process(target=fun1, args=(\"p2\",))\n",
    "\n",
    "    # 启动进程\n",
    "    p1.start()\n",
    "    p2.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b71fba7f",
   "metadata": {},
   "source": [
    "### 类模式 创建进程\n",
    "\n",
    "**Process类的构造方法格式如下：**\n",
    "\n",
    "def __init__(self, group=None, target=None, name=None, args=(), kwargs={})\n",
    "- 其中，各个参数的含义为：\n",
    "- group ：该参数未进行实现，不需要传参；\n",
    "- target ：为新建进程指定执行任务，也就是指定一个函数；\n",
    "- name ：为新建进程设置名称；\n",
    "- args ：为 target 参数指定的参数传递非关键字参数；\n",
    "- kwargs ：为 target 参数指定的参数传递关键字参数。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "91dbd9b2",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "下述代码在 jupyter notebook上 无法正常运行，\n",
    "请使用 Pycharm运行，\n",
    "或者将代码保存为.py文件，再导入jupyter运行\n",
    "\"\"\"\n",
    "from multiprocessing import Process\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "class MyProcess(Process):\n",
    "    \"\"\"继承 Process类\"\"\"\n",
    "\n",
    "    def __init__(self, name):\n",
    "        Process.__init__(self)\n",
    "        self.name = name\n",
    "\n",
    "    def run(self):\n",
    "        \"\"\"进程运行的函数\"\"\"\n",
    "\n",
    "        print(f\"Process:{self.name}, start\")\n",
    "\n",
    "        sleep(1)\n",
    "\n",
    "        print(f\"Process:{self.name}, end\")\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # 创建进程\n",
    "    p1 = MyProcess(\"p1\")\n",
    "    p2 = MyProcess(\"p2\")\n",
    "    \n",
    "    # 启动进程\n",
    "    p1.start()\n",
    "    p2.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d7679863",
   "metadata": {},
   "source": [
    "### Queue实现进程间通信\n",
    "前面讲解了使用 Queue 模块中的 Queue 类实现线程间通信\n",
    "\n",
    "但要实现进程间通信，需要使用 multiprocessing 模块中的 Queue 类\n",
    "\n",
    "Queue 实现进程间通信的方式，就是使用了操作系统给开辟的一个队列空间，各个进程可以把数据放到该队列中，也可以从队列中把自己需要的信息取走。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e8247745",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/14.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8b8d4be2",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "下述代码在 jupyter notebook上 无法正常运行，\n",
    "请使用 Pycharm运行，\n",
    "或者将代码保存为.py文件，再导入jupyter运行\n",
    "\"\"\"\n",
    "from multiprocessing import Process, Queue\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "class MyProcess(Process):\n",
    "    \"\"\"继承Process类\"\"\"\n",
    "    def __init__(self, name, mq):\n",
    "        Process.__init__(self)\n",
    "        self.name = name\n",
    "        self.mq = mq\n",
    "\n",
    "    def run(self):\n",
    "        print(f\"Process:{self.name}, start\")\n",
    "        \n",
    "        # 取出数据\n",
    "        print(f\"get Data:{self.mq.get()}\")\n",
    "        \n",
    "        sleep(1)\n",
    "        \n",
    "        # 存入数据\n",
    "        self.mq.put(f\"new_data:{self.name}\")\n",
    "        \n",
    "        \n",
    "        print(f\"Process:{self.name}, end\")\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    \n",
    "    # 创建一个队列，并存入数据\n",
    "    mq = Queue()\n",
    "    mq.put(\"1\")\n",
    "    mq.put(\"2\")\n",
    "    mq.put(\"3\")\n",
    "\n",
    "    # 创建 3个进程，并放入进程列表\n",
    "    p_list = []\n",
    "    for i in range(3):\n",
    "        \n",
    "        # 将队列 作为参数传入，使得进程可以使用 Queue中的数据\n",
    "        p = MyProcess(f\"p{i}\", mq) \n",
    "        p_list.append(p)\n",
    "\n",
    "    # 运行进程\n",
    "    for p in p_list:\n",
    "        p.start()\n",
    "\n",
    "    # 主进程等待 子进程全部结束，再结束\n",
    "    for p in p_list:\n",
    "        p.join()\n",
    "\n",
    "    # 取出队列中的数据，发现子进程已经把 Queue中的数据替换\n",
    "    print(mq.get())\n",
    "    print(mq.get())\n",
    "    print(mq.get())"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6e82a498",
   "metadata": {},
   "source": [
    "### Pipe实现进程间通信"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "73558e61",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/15.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f5c43052",
   "metadata": {},
   "source": [
    "- Pipe 方法返回（conn1， conn2）代表一个管道的两个端\n",
    "\n",
    "\n",
    "- Pipe方法 有**duplex参数**:\n",
    "    - 如果duplex参数为True（默认值），那么这个参数是全双工模式，也就是说conn1和conn2均可收发。\n",
    "    - 若duplex为False，conn1只负责接收消息，conn2只负责发送消息。\n",
    "    \n",
    "    \n",
    "- **send** 和 **recv方法** 分别是发送和接受消息的方法:\n",
    "    - 例如，在全双工模式下，可以调用conn1.send发送消息，conn1.recv接收消息。\n",
    "    - 如果没有消息可接收，recv方法会一直阻塞。如果管道已经被关闭，那么recv方法会抛出EOFError\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "feb60c14",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "下述代码在 jupyter notebook上 无法正常运行，\n",
    "请使用 Pycharm运行，\n",
    "或者将代码保存为.py文件，再导入jupyter运行\n",
    "\"\"\"\n",
    "import multiprocessing\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "def func1(conn1):\n",
    "    sub_info = \"Hello!\"\n",
    "    \n",
    "    print(f\"进程1--{multiprocessing.current_process().pid}，发送数据：{sub_info}\")\n",
    "    conn1.send(sub_info)\n",
    "    \n",
    "    sleep(1)\n",
    "    \n",
    "    # 如果没有收到消息，则会阻塞\n",
    "    print(f\"来自进程2的数据:{conn1.recv()}\")\n",
    "    sleep(1)\n",
    "\n",
    "\n",
    "def func2(conn2):\n",
    "    sub_info = \"你好!\"\n",
    "    \n",
    "    print(f\"进程2--{multiprocessing.current_process().pid}，发送数据：{sub_info}\")\n",
    "    conn2.send(sub_info)\n",
    "    \n",
    "    sleep(1)\n",
    "    \n",
    "    print(f\"来自进程1的数据:{conn2.recv()}\")\n",
    "    sleep(1)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # 创建管道，返回 两个端口\n",
    "    conn1, conn2 = multiprocessing.Pipe()\n",
    "    \n",
    "    # 创建 两个子进程，分别绑定一个端口\n",
    "    process1 = multiprocessing.Process(target=func1, args=(conn1,))\n",
    "    process2 = multiprocessing.Process(target=func2, args=(conn2,))\n",
    "    \n",
    "    # 启动子进程\n",
    "    process1.start()\n",
    "    process2.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "93151c7e",
   "metadata": {},
   "source": [
    "### Manager管理器\n",
    "管理器提供了一种创建共享数据的方法，从而可以在不同进程中共享。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9a3dbbbf",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "下述代码在 jupyter notebook上 无法正常运行，\n",
    "请使用 Pycharm运行，\n",
    "或者将代码保存为.py文件，再导入jupyter运行\n",
    "\"\"\"\n",
    "from multiprocessing import Process\n",
    "from multiprocessing import Manager\n",
    "\n",
    "\n",
    "def func(name, m_list, m_dict):\n",
    "    \n",
    "    # 添加数据\n",
    "    m_dict['name'] = '北理工'\n",
    "    m_list.append('你好')\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    \n",
    "    # 创建 Manager 管理器\n",
    "    with Manager() as mgr:\n",
    "        \n",
    "        # 创建列表、字典\n",
    "        m_list = mgr.list()\n",
    "        m_dict = mgr.dict()\n",
    "        \n",
    "        m_list.append('Hello!!')\n",
    "        \n",
    "        # 将管理器对象 作为参数传入，使得进程可以使用 Manager中的数据\n",
    "        p1 = Process(target=func, args=('p1', m_list, m_dict))\n",
    "        \n",
    "        # 启动进程\n",
    "        p1.start()\n",
    "        \n",
    "        # 等p1进程结束，主进程继续执行\n",
    "        p1.join()  \n",
    "        \n",
    "        # 查看 经p1进程修改后的数据\n",
    "        print(f\"主进程:{m_list}\")\n",
    "        print(f\"主进程:{m_dict}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a5b1c730",
   "metadata": {},
   "source": [
    "### 进程池（Pool）\n",
    "Python提供了更好的管理多个进程的方式，就是使用进程池。\n",
    "\n",
    "![](01_并发编程_images/16.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5477d75b",
   "metadata": {},
   "source": [
    "进程池可以提供指定数量的进程给用户使用，即当有新的请求提交到进程池中时，如果池未满，则会创建一个新的进程用来执行该请求；\n",
    "\n",
    "反之，如果池中的进程数已经达到规定最大值，那么该请求就会等待，只要池中有进程空闲下来，该请求就能得到执行。\n",
    "\n",
    "**使用进程池的优点：**\n",
    "1. 提高效率，节省开辟进程和开辟内存空间的时间及销毁进程的时间\n",
    "2. 节省内存空间"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1495cebc",
   "metadata": {},
   "source": [
    "![](01_并发编程_images/17.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eb001e0b",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"进程池使用案例\"\"\"\n",
    "\n",
    "\"\"\"\n",
    "下述代码在 jupyter notebook上 无法正常运行，\n",
    "请使用 Pycharm运行，\n",
    "或者将代码保存为.py文件，再导入jupyter运行\n",
    "\"\"\"\n",
    "from multiprocessing import Pool\n",
    "import os\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "def func1(name):\n",
    "    print(f\"当前进程的ID:{os.getpid()}, {name}\")\n",
    "    sleep(2)\n",
    "    return name\n",
    "\n",
    "\n",
    "def func2(args):\n",
    "    print(args)\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    \n",
    "    # 创建进程池，内有 5个进程\n",
    "    pool = Pool(5)\n",
    "\n",
    "    # 异步执行各个事件\n",
    "    pool.apply_async(func=func1, args=('sxt1',), \n",
    "                     callback=func2) # func1执行完之后，func1的返回值作为参数，传入func2中并执行\n",
    "    \n",
    "    pool.apply_async(func=func1, args=('sxt2',), callback=func2)\n",
    "    pool.apply_async(func=func1, args=('sxt3',), callback=func2)\n",
    "    pool.apply_async(func=func1, args=('sxt4',))\n",
    "    pool.apply_async(func=func1, args=('sxt5',))\n",
    "    pool.apply_async(func=func1, args=('sxt6',))\n",
    "    pool.apply_async(func=func1, args=('sxt7',))\n",
    "    pool.apply_async(func=func1, args=('sxt8',))\n",
    "\n",
    "    # 关闭进程池\n",
    "    pool.close()\n",
    "    \n",
    "    # 回收进程池\n",
    "    pool.join()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "18cbc5e7",
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"使用with管理进程池\"\"\"\n",
    "\n",
    "\"\"\"\n",
    "下述代码在 jupyter notebook上 无法正常运行，\n",
    "请使用 Pycharm运行，\n",
    "或者将代码保存为.py文件，再导入jupyter运行\n",
    "\"\"\"\n",
    "from multiprocessing import Pool\n",
    "import os\n",
    "from time import sleep\n",
    "\n",
    "\n",
    "def func1(name):\n",
    "    print(f\"当前进程的ID:{os.getpid()}, {name}\")\n",
    "    sleep(1)\n",
    "    return name\n",
    "\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    \n",
    "    # 创建进程池 -- 使用 with，可以自动销毁进程池\n",
    "    with Pool(5) as pool:\n",
    "        \n",
    "        # 将 迭代对象 作为参数，传入func1中 -- 相当于多次调用 func1\n",
    "        args = pool.map(func1, ('sxt1,', 'sxt2,', 'sxt3,',\n",
    "                        'sxt4,', 'sxt5,', 'sxt6,', 'sxt7,', 'sxt8,'))\n",
    "        for a in args:\n",
    "            print(a)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "8d03abf3",
   "metadata": {},
   "source": [
    "## Python与协程Coroutines"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6a15613a",
   "metadata": {},
   "source": [
    "**协程的核心(控制流的让出和恢复):**\n",
    "\n",
    "1. 每个协程有自己的执行栈，可以保存自己的执行现场\n",
    "\n",
    "2. 可以由用户程序按需创建协程（比如：遇到io操作）协程“主动让出（yield）”执行权时候, 会保存执行现场(保存中断时的寄存器上下文和栈)，然后切换到其他协程。\n",
    "\n",
    "3. 协程恢复执行（resume）时，根据之前保存的执行现场恢复到中断前的状态，继续执行。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "825f9991",
   "metadata": {},
   "source": [
    "### 协程和多线程比较"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "471bdcff",
   "metadata": {},
   "source": [
    "比如，有3个任务需要完成，每个任务都在等待I/O操作时阻塞自身，阻塞在I/O操作上所花费的时间已经用灰色框标示出来了\n",
    "![](01_并发编程_images/18.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b858f6ce",
   "metadata": {},
   "source": [
    "1. 在单线程同步模型中，任务按照顺序执行。如果某个任务因为 I/O 而阻塞，其他所有的任务都必须等待，直到它完成之后它们才能依次执行。\n",
    "\n",
    "\n",
    "2. 多线程版本中，这3个任务分别在独立的线程中执行。这些线程由操作系统来管理，在多处理器系统上可以并行处理，或者在单处理器系统上交错执行。这使得当某个线程阻塞在某个资源的同时，其他线程得以继续执行。\n",
    "\n",
    "\n",
    "3. 协程版本的程序中，3个任务交错执行，但仍然在一个单独的线程控制中。当处理I/O或者其他昂贵的操作时，注册一个回调到事件循环中，然后当I/O操作完成时继续执行。回调描述了该如何处理某个事件。事件循环轮询所有的事件，当事件到来时将它们分配给等待处理事件的回调函数。"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b1bd18c6",
   "metadata": {},
   "source": [
    "**协程的优点：**\n",
    "1. 由于自身带有上下文和栈，无需线程上下文切换的开销，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级；\n",
    "\n",
    "2. 单线程内就可以实现并发的效果，最大限度地利用cpu，且可扩展性高，成本低\n",
    "\n",
    "3. asyncio协程是写爬虫比较好的方式。比多线程和多进程都好，开辟新的线程和进程是非常耗时的。\n",
    "\n",
    "\n",
    "**协程的缺点：**\n",
    "1. 无法利用多核资源：协程的本质是个单线程,它不能同时将 单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "39328430",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-08T03:13:13.893928Z",
     "start_time": "2023-10-08T03:13:07.860984Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "北京：第0次打印啦\n",
      "北京：第1次打印啦\n",
      "北京：第2次打印啦\n",
      "上海：第0次打印了\n",
      "上海：第1次打印了\n",
      "上海：第2次打印了\n",
      "耗时6.023545265197754\n"
     ]
    }
   ],
   "source": [
    "\"\"\"不使用协程执行多个任务\"\"\"\n",
    "import time\n",
    "\n",
    "\n",
    "def func1():\n",
    "    for i in range(3):\n",
    "        print(f'北京：第{i}次打印啦')\n",
    "        time.sleep(1)\n",
    "        \n",
    "    return \"func1执行完毕\"\n",
    "\n",
    "\n",
    "def func2():\n",
    "    for k in range(3):\n",
    "        print(f'上海：第{k}次打印了')\n",
    "        time.sleep(1)\n",
    "        \n",
    "    return \"func2执行完毕\"\n",
    "\n",
    "\n",
    "def main():\n",
    "    func1()\n",
    "    func2()\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    start_time = time.time()\n",
    "    main()\n",
    "    end_time = time.time()\n",
    "    \n",
    "    # 不使用协程，耗时6秒\n",
    "    print(f\"耗时{end_time-start_time}\")  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aa7bc058",
   "metadata": {},
   "source": [
    "### 使用yield实现协程\n",
    "Python中的协程经历了很长的一段发展历程，其大概经历了如下三个阶段：\n",
    "1. 最初的生成器变形 **yield / send**\n",
    "2. 引入 **@asyncio.coroutine** 和 **yield from**\n",
    "3. Python3.5版本后，引入 **async / await** 关键字"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "f6ed0aaa",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-08T03:16:48.702353Z",
     "start_time": "2023-10-08T03:16:43.671398Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'generator'>\n",
      "上海：第0次打印了\n",
      "北京：第0次打印啦\n",
      "上海：第1次打印了\n",
      "北京：第1次打印啦\n",
      "上海：第2次打印了\n",
      "北京：第2次打印啦\n",
      "耗时5.024545192718506\n"
     ]
    }
   ],
   "source": [
    "\"\"\"使用yield实现协程\"\"\"\n",
    "import time\n",
    "\n",
    "\n",
    "def func1():\n",
    "    for i in range(3):\n",
    "        print(f'北京：第{i}次打印啦')\n",
    "        \n",
    "        # 只要方法中包含了yield，func1 就变成一个生成器\n",
    "        yield\n",
    "        \n",
    "        time.sleep(1)\n",
    "\n",
    "\n",
    "def func2():\n",
    "    \n",
    "    # func1是一个生成器，func1()就不会直接调用，需要通过 next()\n",
    "    g = func1()  \n",
    "    print(type(g))\n",
    "    \n",
    "    for k in range(3):\n",
    "        print(f'上海：第{k}次打印了')\n",
    "        \n",
    "        # 继续执行func1 yield之后的代码\n",
    "        next(g)  \n",
    "        time.sleep(1)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    # 有了yield，我们实现了两个任务的切换、保存状态\n",
    "    start_time = time.time()\n",
    "    func2()\n",
    "    end_time = time.time()\n",
    "    \n",
    "    # 耗时5.0秒，效率差别不大 -- yield方式，无法节省 I/O 操作时间\n",
    "    print(f\"耗时{end_time-start_time}\")  "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "79596aea",
   "metadata": {},
   "source": [
    "### asyncio实现协程 (重点) \n",
    "\n",
    "1. 正常的函数执行时是不会中断的，所以你要写一个能够中断的函数，就需要加 async \n",
    "\n",
    "2. async 用来声明一个函数为异步函数，异步函数的特点是能在函数执行过程中挂起，去执行其他异步函数\n",
    "\n",
    "3. 等到挂起条件消失后，假设挂起条件是 sleep(5) ，再回来执行\n",
    "\n",
    "4. await 用来用来声明程序挂起，比如异步程序执行到某一步时需要等待的时间很长，就将此挂起，去执行其他的异步程序。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "fec32ec5",
   "metadata": {
    "ExecuteTime": {
     "end_time": "2023-10-08T03:25:52.462099Z",
     "start_time": "2023-10-08T03:25:49.449824Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "北京：第0次打印啦\n",
      "上海：第0次打印了\n",
      "北京：第1次打印啦\n",
      "上海：第1次打印了\n",
      "北京：第2次打印啦\n",
      "上海：第2次打印了\n",
      "['func1执行完毕', 'func2执行完毕']\n",
      "耗时3.005197048187256\n"
     ]
    }
   ],
   "source": [
    "import asyncio\n",
    "import time\n",
    "\n",
    "\n",
    "async def func1():  \n",
    "    \"\"\"async表示方法是异步的\"\"\"\n",
    "    \n",
    "    for i in range(3):\n",
    "        print(f'北京：第{i}次打印啦')\n",
    "        \n",
    "        # 程序挂起，等待相应事件，此时cpu执行其他事务\n",
    "        await asyncio.sleep(1)\n",
    "        \n",
    "    return \"func1执行完毕\"\n",
    "\n",
    "\n",
    "async def func2():\n",
    "    for k in range(3):\n",
    "        print(f'上海：第{k}次打印了')\n",
    "        \n",
    "        # 程序挂起，等待相应事件，此时cpu执行其他事务\n",
    "        await asyncio.sleep(1)\n",
    "        \n",
    "    return \"func2执行完毕\"\n",
    "\n",
    "\n",
    "async def main():\n",
    "    \n",
    "    # await异步执行func1、func2方法，返回值为函数的返回值列表\n",
    "    res = await asyncio.gather(func1(), func2())\n",
    "    \n",
    "    print(res)\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    start_time = time.time()\n",
    "    \n",
    "    # 异步执行 main函数，jupyter中执行\n",
    "    await main()\n",
    "    \n",
    "    # pycharm中则采用另一种方式执行：asyncio.run(main())\n",
    "    \n",
    "    end_time = time.time()\n",
    "    \n",
    "    # 耗时3秒,效率极大提高\n",
    "    print(f\"耗时{end_time-start_time}\")  "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dde23580",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.12"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "288px"
   },
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
